package com.sugo.seckill.distributedlock.rocketmq;

import com.sugo.seckill.order.service.SeckillOrderService;
import com.sugo.seckill.utils.ObjectUtils;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.io.UnsupportedEncodingException;
import java.nio.charset.Charset;
import java.util.concurrent.*;

/**
 * <p>rocketMq的生产者(发送者)</p>
 *
 * @author langxing
 * @date 2021/1/28 22:02
 */
public class RocketMQSender {

    private static final String producerGroup = "seckillGroup";

    private static final String namesrvAddr = "127.0.0.1:9876";
    private TransactionMQProducer producer;

    /** 获取当前可用的cpu核心数 */
    private static int corePoolSize = Runtime.getRuntime().availableProcessors();
    /** 创建线程池  调整队列数 拒绝服务 */
    private static ExecutorService executor = new ThreadPoolExecutor(corePoolSize,
            corePoolSize + 1,
            10L,
            TimeUnit.SECONDS,
            new ArrayBlockingQueue<>(1000),
            r -> new Thread(r, "RocketMQ-thread"));

    @Autowired
    SeckillOrderService seckillOrderService;

    @PostConstruct
    public void initProducer() {
        producer = new TransactionMQProducer(producerGroup);
        producer.setNamesrvAddr(namesrvAddr);
        producer.setRetryTimesWhenSendFailed(3);
        try {
            // 设置线程池
            producer.setExecutorService(executor);
            producer.start();
            // 注册生产者监听器
            producer.setTransactionListener(new TransactionListener() {
                /**
                 * 本地事务执行方法
                 * @param message
                 * @param o
                 * @return
                 */
                @Override
                public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                    try {
                        String seckillId = new String(message.getBody(), RemotingHelper.DEFAULT_CHARSET);
                        String userId = ObjectUtils.toString(o);
                        // 执行业务
                        seckillOrderService.placeOrder(Long.parseLong(seckillId), userId);
                    } catch (UnsupportedEncodingException e) {
                        e.printStackTrace();
                        return LocalTransactionState.ROLLBACK_MESSAGE;
                    }
                    return LocalTransactionState.COMMIT_MESSAGE;
                }

                /**
                 * 本地事务状态回查方法
                 * @param messageExt
                 * @return
                 */
                @Override
                public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                    System.out.println("服务器端回查事务消息： "+messageExt.toString());
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
            });
            System.out.println("[Producer 已启动]");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @PreDestroy
    public void shutDownProducer() {
        if (producer != null) {
            producer.shutdown();
        }
    }

    /**
     * 事务+异步发送MQ信息
     * @param topicName
     * @param seckillId
     * @param userId
     * @author linhx
     * @date 2021/1/31 14:31
     * @return boolean
     */
    public boolean transactionAsyncReduceStock(String topicName, String seckillId, String userId) {
        Message message = new Message(topicName, "stock_plus", seckillId.getBytes(Charset.forName("UTF-8")));
        TransactionSendResult transactionSendResult;
        try {
            transactionSendResult = producer.sendMessageInTransaction(message, userId);
        } catch (MQClientException e) {
            e.printStackTrace();
            return false;
        }
        // 判断 transactionSendResult 的状态
        switch (transactionSendResult.getLocalTransactionState()) {
            case ROLLBACK_MESSAGE:
                // 回滚信息
                return false;
            case COMMIT_MESSAGE:
                // 成功提交
                return true;
            default:
                // 其他信息
                return false;
        }
    }

    public String send(String topic, String tags, String msg) {
        SendResult result = null;
        try {
            Message message = new Message(topic, tags, msg.getBytes(RemotingHelper.DEFAULT_CHARSET));
            result = producer.send(message);
            System.out.println("[Producer] msgID(" + result.getMsgId() + ") " + result.getSendStatus());
        } catch (Exception e) {
            e.printStackTrace();
        }
        return "{\"MsgId\":\"" + result.getMsgId() + "\"}";
    }

    /**
     * 同步库存扣减消息
     * @param topicName
     * @param seckillId
     * @return
     */
    public boolean asyncReduceStock(String topicName, String seckillId) {
        Message message = new Message(topicName, "stock_plus", seckillId.getBytes(Charset.forName("UTF-8")));
        try {
            producer.send(message);
        } catch (MQClientException e) {
            e.printStackTrace();
            return false;
        } catch (RemotingException e) {
            e.printStackTrace();
            return false;
        } catch (MQBrokerException e) {
            e.printStackTrace();
            return false;
        } catch (InterruptedException e) {
            e.printStackTrace();
            return false;
        }
        return true;
    }
}
